package com.xiongjie.SingleLimit.Interface.impl;

import com.google.common.util.concurrent.RateLimiter;
import com.xiongjie.SingleLimit.Interface.ISingleLimit;
import org.infinispan.Cache;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public abstract class ISingleLimitImpl implements ISingleLimit, Runnable {

    private final ConcurrentLinkedQueue<String> container = new ConcurrentLinkedQueue<>();
    private ReentrantLock lock = new ReentrantLock(true);   //并发锁


    private AtomicInteger counter;   //限流的核心是整体访问数量的限制,分布式时，也可以将计数器存储到infinispan中
    final private Cache<String, String> cache;

    private Long lastVisitAt = System.currentTimeMillis();  //作为变量锁，同步操作，同时定时清空限流计数器
    private final int block = 10;    //窗口上限
    private final AtomicInteger[] countPerBlock;    //窗口，每个记录有访问次数
    private volatile int index = 0;     //原子移动下标

    private RateLimiter rateLimiter = RateLimiter.create(10);


    public ISingleLimitImpl(Cache<String, String> cache) {
        this.cache = cache;
        counter = new AtomicInteger(0);
        countPerBlock = new AtomicInteger[block];
        for (int i = 0; i < block; i++) {
            countPerBlock[i] = new AtomicInteger(0);
        }

        //滑动窗口+定时器限流启用
//        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
//        scheduledExecutorService.scheduleAtFixedRate(this, 100, 75, TimeUnit.MILLISECONDS);

        //漏桶算法+定时器启用
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(this, 100, 75, TimeUnit.MILLISECONDS);
    }

    protected abstract String getSingleLimitValue(String name);

    @Override
    public void run() {

//        //滑动窗口代码
//        index = (index + 1) % block;
//        int val = countPerBlock[index].getAndSet(0);
//        System.out.println(String.format("当前Qps为[%s]新窗口[%s]原本的值为[%s]",counter.get(),index,val));
//        counter.addAndGet(-val);

        //漏桶限流代码-->定时取数据进行执行，逻辑不能够立即返回给对应的值，所以不推荐
        String msg;
        if (!container.isEmpty()) {
            rateLimiter.acquire();
            String name = container.poll();    //从队列取出请求处理

            //开始业务访问
            String value = cache.get(name);
            if (value != null) {
                msg = "访问成功！返回结果为缓存：" + value;
                System.out.println(msg);
            } else {
                String res = getSingleLimitValue(name);
                cache.put(name, res);
                msg = "访问成功！返回结果为调用值：" + res;
            }
        } else {
            //当木桶的消费完后，可以消费那些降级存入MQ或者DB里面的数据
            System.out.println("将会消费降级后的数据，从消息队列或者数据库中");
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    /**
     * 限流代码的核心,这里展示几种限流算法
     *
     * @param name
     * @return
     */
    @Override
    final public String getValueByName(String name) {
        String msg;

//      msg=simpleCounterLimit(name);
//      msg = fixedWindowLimit(name);
//      msg = slidingWindowLimitWithQuartz(name);
//      msg=slidingWindowLimitWithTimeVar(name);

        msg = leadkyBucketLimit(name);

//      msg=tokenBucketLimit(name);

        return msg;
    }

    /**
     * 使用简单计数器进行限流，本质是到达指定次数就立刻丢弃后续的请求，直到计数器清零
     * <p>
     * 计数器其实就是固定窗口，我写的这个方法实际不算限流算法，因为没有时间范围代码
     *
     * @param name
     * @return
     */
    private String simpleCounterLimit(String name) {
        String msg;

        String value = cache.get(name);
        counter.incrementAndGet();

        //开始限流
        if (counter.get() > 5) {
            counter.set(0);
            msg = "请求数大于阈值，访问失败！";

        } else {

            //开始缓存
            if (value != null) {
                msg = "访问成功！返回结果为缓存：" + value;
            } else {
                String res = getSingleLimitValue(name);
                cache.put(name, res);
                msg = "访问成功！返回结果为调用值：" + res;
            }

        }
        return msg;
    }

    /**
     * 固定窗口限流，需要和一个定时任务搭配，将一个全局变量或者说缓存数据清空为0
     * 本质是划分时间片，需要使用定时任务或者新开一个线程，或者时间间隔变量
     * <p>
     * 比如某个API在一分钟内只能被固定访问N次
     * 该算法的致命问题是临界问题：用户在重置节点的时候突发大量请求，可能导致系统奔溃
     * <p>
     * 只有加法和重置
     *
     * @param name
     * @return
     */
    private String fixedWindowLimit(String name) {

        long now = System.currentTimeMillis();
        synchronized (lastVisitAt) {
            if (now - lastVisitAt > 1000) {
                lastVisitAt = now;
                counter.set(0);
            }
        }

        String msg;
        counter.incrementAndGet();
        if (counter.get() > 5) {
            msg = "请求数大于阈值，访问失败！";
            return msg;
        }

        //开始业务访问
        String value = cache.get(name);
        if (value != null) {
            msg = "访问成功！返回结果为缓存：" + value;
        } else {
            String res = getSingleLimitValue(name);
            cache.put(name, res);
            msg = "访问成功！返回结果为调用值：" + res;
        }

        return msg;
    }

    /**
     * 滑动窗口限流+使用定时器清空计数器
     * <p>
     * 有加法和减法
     *
     * @param name
     * @return
     */
    private String slidingWindowLimitWithQuartz(String name) {
        String msg;

        countPerBlock[index].incrementAndGet();
        counter.incrementAndGet();
        if (counter.get() > 20) {
            msg = "请求数大于阈值，访问失败！";
            return msg;
        }

        String value = cache.get(name);
        if (value != null) {
            msg = "访问成功！返回结果为缓存：" + value;
        } else {
            String res = getSingleLimitValue(name);
            cache.put(name, res);
            msg = "访问成功！返回结果为调用值：" + res;
        }

        return msg;
    }

    /**
     * 滑动窗口限流+使用时间变量清空计数器
     *
     * @param name
     * @return
     */
    private String slidingWindowLimitWithTimeVar(String name) {

        long now = System.currentTimeMillis();
        synchronized (lastVisitAt) {
            if (now - lastVisitAt > 1000) {
                index = (index + 1) % block;
                int val = countPerBlock[index].getAndSet(0);
                System.out.println(String.format("当前Qps为[%s]新窗口[%s]原本的值为[%s]", counter.get(), index, val));
                counter.addAndGet(-val);
            }
        }

        String msg;
        countPerBlock[index].incrementAndGet();
        counter.incrementAndGet();
        if (counter.get() > 20) {
            msg = "请求数大于阈值，访问失败！";
            return msg;
        }

        String value = cache.get(name);
        if (value != null) {
            msg = "访问成功！返回结果为缓存：" + value;
        } else {
            String res = getSingleLimitValue(name);
            cache.put(name, res);
            msg = "访问成功！返回结果为调用值：" + res;
        }

        return msg;
    }

    /**
     * 使用漏桶算法实现限流算法。本质是令牌桶的反向操作，定时的将作为漏桶的全局变量或者阻塞队列进行减法操作。
     * 只有当每个请求在漏桶中时，该请求才会继续执行，请求过多导致漏桶溢出，请求丢弃
     * <p>
     * 效果是在桶的大小恒定的情况下，指定的时间范围只能多少个请求
     * 有加法和减法
     *
     * @param name
     * @return
     */
    private String leadkyBucketLimit(String name) {
        String msg;

        final ReentrantLock lock = this.lock;
        lock.lock();

        try {
            if (container.size() < 50) {
                container.offer(name);
                msg = "提交数据：" + name + "队列大小=" + container.size();
            } else {
                //这里时候采用降级策略了。消费速度跟不上产生速度时，而且桶满了，抛出异常，或者存入MQ DB等后续处理
                msg = "漏桶已满，请稍后再试";
            }
        } finally {
            lock.unlock();
        }

        return msg;
    }

    /**
     * 使用令牌桶实现限流算法。本质是定时往阻塞队列(令牌桶)放令牌，每个请求来时，则从阻塞队列获取令牌，有则继续访问，无则丢弃
     * <p>
     * 当然，这个令牌桶也可以使用全局变量或者共享变量来充当。关键是有一个定时任务不同的操作该令牌桶
     * 有加法和减法
     *
     * @param name
     * @return
     */
    private String tokenBucketLimit(String name) {
        String msg;

        boolean limit = rateLimiter.tryAcquire(1, 100, TimeUnit.MILLISECONDS);
        if (!limit) {
            msg = "请求数大于阈值，访问失败！";
            return msg;
        }

        String value = cache.get(name);
        if (value != null) {
            msg = "访问成功！返回结果为缓存：" + value;
        } else {
            String res = getSingleLimitValue(name);
            cache.put(name, res);
            msg = "访问成功！返回结果为调用值：" + res;
        }

        return msg;
    }

}
